/*!
 * \file ems_message.c
 *
 * \brief 主要对应TibjmsMessage类
 *
 * All Rights Reserved Copyright (C) 2014 broada,Co.,Ltd.
 *
 * \version 1.0
 * \author songbb
 * \date 2016/10/24
 *
 */

#include "conntrack.h"
#include "frct_comm.h"

#include "ems_core.h"
#include "ems_comm.h"
#include "ems_message.h"
#include "ems_xstream.h"


int ems_readHeaders(Conntrack *p_st_conn
                    , L7_ems_packet *p_st_ems_packet
                    , int *p_i_compressed
                    , Frreport_body_data *p_st_body)
{
    EMS_PRINT_INFO("ems_readHeaders...\n");
    L7_session *p_st_session = frct_get_session(p_st_conn);
    L7session_ems_private *p_st_session_private = (L7session_ems_private *) frct_get_session_private(p_st_session);
    L7session_ems *p_st_sess = p_st_session_private->p_st_sess;

    unsigned short us_header_size;
    char c_jmspriority;
    char c_jmsdeliverymode;
    unsigned char uc_jmsredelivered;
    long long ll_sessid;
    char *p_jmsdestination = NULL;
    int i_jmsdestination_len;
    long long ll_jmsexpiration;
    int j;
    unsigned char uc_k;
    long long ll_jmstimestamp;
    char *p_jmstype = NULL;
    int i_jmstype_len;
    char *p_jmsmessageid = NULL;
    int i_jmsmessageid_len;
    char *p_jmscorrelationid = NULL;
    int i_jmscorrelationid_len;
    char *p_jmsreplyto = NULL;
    int i_jmsreplyto_len;
    char c_compressed;
    long long ll_deliverycount = 0LL;
    char *p_encoding = NULL;
    int i_encoding_len;
    Ems_object st_object_type;
    memset(&st_object_type, 0x0, sizeof(Ems_object));
    char *p_title = NULL;
    int i_title_len = 0;

    if (ems_readUnsignedShort(p_st_ems_packet, &us_header_size) != FR_SUCC)
        goto error;

    j = bytebuff_getPosition(p_st_ems_packet->p_st_packet);

    if (ems_readByte(p_st_ems_packet, &c_jmspriority) != FR_SUCC)
        goto error;

    if (ems_readByte(p_st_ems_packet, &c_jmsdeliverymode) != FR_SUCC)
        goto error;

    if (ems_readUnsignedByte(p_st_ems_packet, &uc_jmsredelivered) != FR_SUCC)
        goto error;

    if (ems_readVarLong(p_st_ems_packet, &ll_sessid) != FR_SUCC)
        goto error;

    if (ems_readDestination(p_st_sess, p_st_ems_packet, &p_jmsdestination, &i_jmsdestination_len) != FR_SUCC)
        goto error;
    if (p_jmsdestination && p_st_sess->model_type != EMS_MODEL_TYPE_UNKNOWN)
    {
        EMS_PRINT_DBG("_JMSDestination:[%s]\n", p_jmsdestination);
        if (p_st_body)
        {
            if (frreport_body_l7_add_str(p_st_body, "tibcoems.headers.jmsdestination", p_jmsdestination) != FR_SUCC)
                goto error;
            i_title_len = strlen(p_st_sess->model_type == EMS_MODEL_TYPE_PRODUCER ? "GET " : "NOTIFY ") + strlen(p_jmsdestination) + 1;
            if ((p_title = (char *) frmalloc(i_title_len * sizeof(char))) == NULL)
            {
                EMS_MALLOCK_ERR;
                goto error;
            }
            snprintf(p_title, i_title_len, p_st_sess->model_type == EMS_MODEL_TYPE_PRODUCER ? "GET %s" : "NOTIFY %s", p_jmsdestination);
            if (frreport_body_l7_add_str(p_st_body, APM_REPORT_KEY_TRANS_TITLE, p_title) != FR_SUCC)
                goto error;
        }
    }

    if (ems_readVarLong(p_st_ems_packet, &ll_jmsexpiration) != FR_SUCC)
        goto error;
    EMS_PRINT_DBG("_JMSExpiration:[%lld]\n", ll_jmsexpiration);

    while (bytebuff_getPosition(p_st_ems_packet->p_st_packet) < us_header_size + j)
    {
        if (ems_readUnsignedByte(p_st_ems_packet, &uc_k) != FR_SUCC)
            goto error;
        switch (uc_k)
        {
            case 62:    // _PRODUCER_ID_
                EMS_PRINT_CRIT("_PRODUCER_ID_\n");
                if (ems_readObject(p_st_sess, p_st_ems_packet, &st_object_type) != FR_SUCC)
                    goto error;
                switch (st_object_type.type)
                {
                    case EMS_OBJECT_TYPE_NULL: // 0
                        EMS_PRINT_DBG("_PRODUCER_ID_0:[NULL]\n");
                        if (p_st_body && frreport_body_l7_add_str(p_st_body, "tibcoems.headers.producer_id", "") != FR_SUCC)
                            goto error;
                        break;
                    case EMS_OBJECT_TYPE_BOOL: // 1
                        EMS_PRINT_DBG("_PRODUCER_ID_1:[%s]\n", st_object_type.value.c_bool == FR_YES ? "YES" : "NO");
                        if (p_st_body && frreport_body_l7_add_str(p_st_body, "tibcoems.headers.producer_id", st_object_type.value.c_bool == FR_YES ? "YES" : "NO") != FR_SUCC)
                            goto error;
                        break;
                    case EMS_OBJECT_TYPE_BYTE: // 2
                        EMS_PRINT_DBG("_PRODUCER_ID_2:[%d]\n", st_object_type.value.c_byte);
                        if (p_st_body && frreport_body_l7_add_int(p_st_body, "tibcoems.headers.producer_id", st_object_type.value.c_byte) != FR_SUCC)
                            goto error;
                        break;
                    case EMS_OBJECT_TYPE_WCHAR: // 3
                        EMS_PRINT_DBG("_PRODUCER_ID_3:[%d]\n", st_object_type.value.s_wchar);
                        if (p_st_body && frreport_body_l7_add_int(p_st_body, "tibcoems.headers.producer_id", st_object_type.value.s_wchar) != FR_SUCC)
                            goto error;
                        break;
                    case EMS_OBJECT_TYPE_SHORT: // 4
                        EMS_PRINT_DBG("_PRODUCER_ID_4:[%d]\n", st_object_type.value.s_short);
                        if (p_st_body && frreport_body_l7_add_int(p_st_body, "tibcoems.headers.producer_id", st_object_type.value.s_short) != FR_SUCC)
                            goto error;
                        break;
                    case EMS_OBJECT_TYPE_INT: // 5
                        EMS_PRINT_DBG("_PRODUCER_ID_5:[%d]\n", st_object_type.value.i_int);
                        if (p_st_body && frreport_body_l7_add_int(p_st_body, "tibcoems.headers.producer_id", st_object_type.value.i_int) != FR_SUCC)
                            goto error;
                        break;
                    case EMS_OBJECT_TYPE_LONG: // 6
                        EMS_PRINT_DBG("_PRODUCER_ID_6:[%lld]\n", st_object_type.value.ll_long);
                        if (p_st_body && frreport_body_l7_add_long(p_st_body, "tibcoems.headers.producer_id", st_object_type.value.ll_long) != FR_SUCC)
                            goto error;
                        break;
                    case EMS_OBJECT_TYPE_FLOAT: // 7
                        EMS_PRINT_DBG("_PRODUCER_ID_7:[%f]\n", st_object_type.value.f_float);
                        if (p_st_body && frreport_body_l7_add_float(p_st_body, "tibcoems.headers.producer_id", st_object_type.value.f_float) != FR_SUCC)
                            goto error;
                        break;
                    case EMS_OBJECT_TYPE_DOUBLE: // 8
                        EMS_PRINT_DBG("_PRODUCER_ID_8:[%f]\n", st_object_type.value.d_double);
                        if (p_st_body && frreport_body_l7_add_double(p_st_body, "tibcoems.headers.producer_id", st_object_type.value.d_double) != FR_SUCC)
                            goto error;
                        break;
                    case EMS_OBJECT_TYPE_UTF8: // 9
                        EMS_PRINT_DBG("_PRODUCER_ID_9:[%s]\n", st_object_type.value.p_utf8);
                        if (p_st_body && frreport_body_l7_add_str(p_st_body, "tibcoems.headers.producer_id", st_object_type.value.p_utf8) != FR_SUCC)
                            goto error;
                        ems_free((void **) &(st_object_type.value.p_utf8));
                        break;
                    case EMS_OBJECT_TYPE_BYTES: // 10
                    case EMS_OBJECT_TYPE_SUBMAP: // 11
                    case EMS_OBJECT_TYPE_SUBSTREAM: // 18
                    case EMS_OBJECT_TYPE_SHORTA_ARRAY: // 12
                    case EMS_OBJECT_TYPE_INT_ARRAY: // 13
                    case EMS_OBJECT_TYPE_LONG_ARRAY: // 14
                    case EMS_OBJECT_TYPE_FLOAT_ARRAY: // 15
                    case EMS_OBJECT_TYPE_DOUBLE_ARRAY: // 16
                        EMS_PRINT_DBG("_PRODUCER_ID_:[NOT SUPPORT]\n");
                        if (p_st_body && frreport_body_l7_add_str(p_st_body, "tibcoems.headers.producer_id", "INTERNAL NOT SUPPORT") != FR_SUCC)
                            goto error;
                        break;
                    case EMS_OBJECT_TYPE_VARLONG: // 17
                        EMS_PRINT_DBG("_PRODUCER_ID_:[%lld]\n", st_object_type.value.ll_varlong);
                        if (p_st_body && frreport_body_l7_add_long(p_st_body, "tibcoems.headers.producer_id", st_object_type.value.ll_varlong) != FR_SUCC)
                            goto error;
                        break;
                    default:
                        goto error;
                }
                break;
            case 50:    // _TIMESTAMP_
                EMS_PRINT_CRIT("_TIMESTAMP_\n");
                if (ems_readLong(p_st_ems_packet, &ll_jmstimestamp) != FR_SUCC)
                    goto error;
                break;
            case 53:    // _MSGTYPE_
                EMS_PRINT_CRIT("_MSGTYPE_\n");
                if (ems_readString(p_st_sess, p_st_ems_packet, FR_NO, &p_jmstype, &i_jmstype_len) != FR_SUCC)
                    goto error;
                break;
            case 54:    // _MSGID_
                EMS_PRINT_CRIT("_MSGID_\n");
                if (ems_readString(p_st_sess, p_st_ems_packet, FR_NO, &p_jmsmessageid, &i_jmsmessageid_len) != FR_SUCC)
                    goto error;
                EMS_PRINT_DBG("_JMSMessageID:[%s]\n", p_jmsmessageid);
                if (p_st_body && p_jmsmessageid && frreport_body_l7_add_str(p_st_body, "tibcoems.headers.jmsmessageid", p_jmsmessageid) != FR_SUCC)
                    goto error;
                break;
            case 55:    // _CORID_
                EMS_PRINT_CRIT("_CORID_\n");
                if (ems_readString(p_st_sess, p_st_ems_packet, FR_NO, &p_jmscorrelationid, &i_jmscorrelationid_len) != FR_SUCC)
                    goto error;
                break;
            case 52:    // _REPLY_
                EMS_PRINT_CRIT("_REPLY_\n");
                if (ems_readDestination(p_st_sess, p_st_ems_packet, &p_jmsreplyto, &i_jmsreplyto_len) != FR_SUCC)
                    goto error;
                break;
            case 61:    // _COMPRESSED_
                EMS_PRINT_CRIT("_COMPRESSED_\n");
                if (ems_readByte(p_st_ems_packet, &c_compressed) != FR_SUCC)
                    goto error;
                *p_i_compressed = c_compressed == 1 ? FR_YES : FR_NO;
                EMS_PRINT_DBG("compressed:[%s]\n", c_compressed == 1 ? "YES" : "NO");
                break;
            case 64:    // _DELIVERY_COUNT_
                EMS_PRINT_CRIT("_DELIVERY_COUNT_\n");
                if (ems_readTypedLong(p_st_ems_packet, &ll_deliverycount) != FR_SUCC)
                    goto error;
                break;
            case 65:    // _ENCODING_
                EMS_PRINT_CRIT("_ENCODING_\n");
                if (ems_readString(p_st_sess, p_st_ems_packet, FR_NO, &p_encoding, &i_encoding_len) != FR_SUCC)
                    goto error;
                EMS_PRINT_DBG("_encoding:[%s]\n", p_encoding);
                break;
            case 51:    // _REDELIVERED_
            case 56:
            case 57:
            case 58:
            case 59:
            case 60:    // _TXID_
            case 63:    // _CLIENT_MSG_FLAGS_
            default:
                EMS_PRINT_CRIT("_other_\n");
                if (ems_skipObject(p_st_sess, p_st_ems_packet) != FR_SUCC)
                    goto error;
        }
    }

    ems_free((void **) &p_title);
    ems_free((void **) &p_jmsdestination);
    ems_free((void **) &p_jmstype);
    ems_free((void **) &p_jmsmessageid);
    ems_free((void **) &p_jmscorrelationid);
    ems_free((void **) &p_jmsreplyto);
    ems_free((void **) &p_encoding);
    return FR_SUCC;
error:
    ems_free((void **) &p_title);
    ems_free((void **) &p_jmsdestination);
    ems_free((void **) &p_jmstype);
    ems_free((void **) &p_jmsmessageid);
    ems_free((void **) &p_jmscorrelationid);
    ems_free((void **) &p_jmsreplyto);
    ems_free((void **) &p_encoding);
    return FR_FAIL;
}

int ems_readProperties(Conntrack *p_st_conn
                    , L7_ems_packet *p_st_ems_packet
                    , Frreport_body_data *p_st_body)
{
    EMS_PRINT_INFO("ems_readProperties...\n");
    L7_session *p_st_session = frct_get_session(p_st_conn);
    L7session_ems_private *p_st_session_private = (L7session_ems_private *) frct_get_session_private(p_st_session);
    L7session_ems *p_st_sess = p_st_session_private->p_st_sess;
    int i_properties_size;
    int j;
    char *p_name = NULL;
    int i_name_len;
    Ems_object st_object_type;
    memset(&st_object_type, 0x0, sizeof(Ems_object));
    int i_report_key_len;
    char *p_report_key = NULL;

    if (ems_readInt(p_st_ems_packet, &i_properties_size) != FR_SUCC)
        goto error;

    if (i_properties_size == 0)
        return FR_SUCC;

    j = bytebuff_getPosition(p_st_ems_packet->p_st_packet);

    while (bytebuff_getPosition(p_st_ems_packet->p_st_packet) < i_properties_size + j)
    {
        if (ems_readString(p_st_sess, p_st_ems_packet, FR_NO, &p_name, &i_name_len) != FR_SUCC)
            goto error;
        if (ems_readObject(p_st_sess, p_st_ems_packet, &st_object_type) != FR_SUCC)
            goto error;

        if (frstrlower(p_name, i_name_len) != FR_SUCC)
            goto error;
        i_report_key_len = (int) (strlen("tibcoems.properties") + i_name_len + 1);
        if ((p_report_key = (char *) frmalloc(sizeof(char) * i_report_key_len)) == NULL)
        {
            EMS_MALLOCK_ERR;
            goto error;
        }
        snprintf(p_report_key, (size_t) i_report_key_len, "tibcoems.properties.%s", p_name);

        switch (st_object_type.type)
        {
            case EMS_OBJECT_TYPE_NULL: // 0
                EMS_PRINT_WARN("property name:[%s] type:[%s] value:[NULL]\n", p_name, EMS_OBJECT_TYPE2STR(st_object_type.type));
                if (p_st_body && frreport_body_l7_add_str(p_st_body, p_report_key, "") != FR_SUCC)
                    goto error;
                break;
            case EMS_OBJECT_TYPE_BOOL: // 1
                EMS_PRINT_WARN("property name:[%s] type:[%s] value:[%s]\n", p_name, EMS_OBJECT_TYPE2STR(st_object_type.type),
                               st_object_type.value.c_bool == FR_YES ? "YES" : "NO");
                if (p_st_body && frreport_body_l7_add_str(p_st_body, p_report_key, st_object_type.value.c_bool == FR_YES ? "YES" : "NO") != FR_SUCC)
                    goto error;
                break;
            case EMS_OBJECT_TYPE_BYTE: // 2
                EMS_PRINT_WARN("property name:[%s] type:[%s] value:[%d]\n", p_name, EMS_OBJECT_TYPE2STR(st_object_type.type),
                               st_object_type.value.c_byte);
                if (p_st_body && frreport_body_l7_add_int(p_st_body, p_report_key, st_object_type.value.c_byte) != FR_SUCC)
                    goto error;
                break;
            case EMS_OBJECT_TYPE_WCHAR: // 3
                EMS_PRINT_WARN("property name:[%s] type:[%s] value:[%d]\n", p_name, EMS_OBJECT_TYPE2STR(st_object_type.type),
                               st_object_type.value.s_wchar);
                if (p_st_body && frreport_body_l7_add_int(p_st_body, p_report_key, st_object_type.value.s_wchar) != FR_SUCC)
                    goto error;
                break;
            case EMS_OBJECT_TYPE_SHORT: // 4
                EMS_PRINT_WARN("property name:[%s] type:[%s] value:[%d]\n", p_name, EMS_OBJECT_TYPE2STR(st_object_type.type),
                               st_object_type.value.s_short);
                if (p_st_body && frreport_body_l7_add_int(p_st_body, p_report_key, st_object_type.value.s_short) != FR_SUCC)
                    goto error;
                break;
            case EMS_OBJECT_TYPE_INT: // 5
                EMS_PRINT_WARN("property name:[%s] type:[%s] value:[%d]\n", p_name, EMS_OBJECT_TYPE2STR(st_object_type.type),
                               st_object_type.value.i_int);
                if (p_st_body && frreport_body_l7_add_int(p_st_body, p_report_key, st_object_type.value.i_int) != FR_SUCC)
                    goto error;
                break;
            case EMS_OBJECT_TYPE_LONG: // 6
                EMS_PRINT_WARN("property name:[%s] type:[%s] value:[%lld]\n", p_name, EMS_OBJECT_TYPE2STR(st_object_type.type),
                               st_object_type.value.ll_long);
                if (p_st_body && frreport_body_l7_add_long(p_st_body, p_report_key, st_object_type.value.ll_long) != FR_SUCC)
                    goto error;
                break;
            case EMS_OBJECT_TYPE_FLOAT: // 7
                EMS_PRINT_WARN("property name:[%s] type:[%s] value:[%f]\n", p_name, EMS_OBJECT_TYPE2STR(st_object_type.type),
                               st_object_type.value.f_float);
                if (p_st_body && frreport_body_l7_add_float(p_st_body, p_report_key, st_object_type.value.f_float) != FR_SUCC)
                    goto error;
                break;
            case EMS_OBJECT_TYPE_DOUBLE: // 8
                EMS_PRINT_WARN("property name:[%s] type:[%s] value:[%f]\n", p_name, EMS_OBJECT_TYPE2STR(st_object_type.type),
                               st_object_type.value.d_double);
                if (p_st_body && frreport_body_l7_add_double(p_st_body, p_report_key, st_object_type.value.d_double) != FR_SUCC)
                    goto error;
                break;
            case EMS_OBJECT_TYPE_UTF8: // 9
                EMS_PRINT_WARN("property name:[%s] type:[%s] value:[%s]\n", p_name, EMS_OBJECT_TYPE2STR(st_object_type.type),
                               st_object_type.value.p_utf8);
                if (p_st_body && frreport_body_l7_add_str(p_st_body, p_report_key, st_object_type.value.p_utf8) != FR_SUCC)
                    goto error;
                ems_free((void **) &(st_object_type.value.p_utf8));
                break;
            case EMS_OBJECT_TYPE_BYTES: // 10
            case EMS_OBJECT_TYPE_SUBMAP: // 11
            case EMS_OBJECT_TYPE_SUBSTREAM: // 18
            case EMS_OBJECT_TYPE_SHORTA_ARRAY: // 12
            case EMS_OBJECT_TYPE_INT_ARRAY: // 13
            case EMS_OBJECT_TYPE_LONG_ARRAY: // 14
            case EMS_OBJECT_TYPE_FLOAT_ARRAY: // 15
            case EMS_OBJECT_TYPE_DOUBLE_ARRAY: // 16
                EMS_PRINT_WARN("property name:[%s] type:[%s] value:[NOT SUPPORT]\n", p_name, EMS_OBJECT_TYPE2STR(st_object_type.type));
                if (p_st_body && frreport_body_l7_add_str(p_st_body, p_report_key, "INTERNAL NOT SUPPORT") != FR_SUCC)
                    goto error;
                break;
            case EMS_OBJECT_TYPE_VARLONG: // 17
                EMS_PRINT_WARN("property name:[%s] type:[%s] value:[%lld]\n", p_name, EMS_OBJECT_TYPE2STR(st_object_type.type),
                               st_object_type.value.ll_varlong);
                if (p_st_body && frreport_body_l7_add_long(p_st_body, p_report_key, st_object_type.value.ll_varlong) != FR_SUCC)
                    goto error;
                break;
            default:
                goto error;
        }

        ems_free((void **) &p_name);
        ems_free((void **) &p_report_key);
    }

    return FR_SUCC;
error:
    ems_free((void **) &p_name);
    ems_free((void **) &p_report_key);
    return FR_FAIL;
}
